package com.ktvme.controller;

import com.ktvme.annotation.OperationLog;
import com.ktvme.core.AbstractFlinkJobService;
import com.ktvme.core.FlinkJobServiceFactory;
import com.ktvme.core.JobInfo;
import com.ktvme.dto.FlinkJobConfigDTO;
import com.ktvme.entity.Cluster;
import com.ktvme.entity.FlinkJobConfig;
import com.ktvme.result.Result;
import com.ktvme.service.FlinkJobConfigService;
import com.ktvme.service.ClusterService;
import com.ktvme.vo.FlinkJobConfigVO;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/flink/configs")
@Tag(name = "Flink任务配置相关接口")
@Slf4j
public class FlinkJobConfigController {

    @Autowired
    private FlinkJobConfigService flinkJobConfigService;
    
    @Autowired
    private ClusterService clusterService;

    private AbstractFlinkJobService getFlinkClient(String clusterId) {
        Cluster cluster = clusterService.getById(clusterId);
        if (cluster == null) {
            throw new IllegalArgumentException("Cluster not found with id: " + clusterId);
        }
        return FlinkJobServiceFactory.createFlinkJobService(cluster.getVersion(), cluster.getUrl());
    }

    @GetMapping("/{clusterId}")
    @Operation(summary = "查询所有Flink任务配置")
    @OperationLog(operationType = "QUERY", description = "查询Flink任务配置列表")
    public Result<List<FlinkJobConfigVO>> list(@PathVariable String clusterId) {
        // 获取所有配置
        List<FlinkJobConfig> configs = flinkJobConfigService.list();
        
        // 获取当前运行的作业
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        List<JobInfo> runningJobs = flinkClient.getRunningJobs();
        Map<String, JobInfo> jobMap = runningJobs.stream()
            .collect(Collectors.toMap(JobInfo::getJobName, job -> job));
        
        // 转换为VO并设置状态
        List<FlinkJobConfigVO> voList = configs.stream().map(config -> {
            FlinkJobConfigVO vo = new FlinkJobConfigVO();
            BeanUtils.copyProperties(config, vo);
            
            // 根据jobName判断状态
            JobInfo job = jobMap.get(config.getJobName());
            if (job != null) {
                vo.setStatus(1); // 运行中
            } else {
                vo.setStatus(0); // 停止
            }
            
            return vo;
        }).collect(Collectors.toList());
        
        return Result.success(voList);
    }

    @GetMapping("/{clusterId}/{id}")
    @Operation(summary = "根据ID查询Flink任务配置")
    @OperationLog(operationType = "QUERY", description = "查询Flink任务配置详情")
    public Result<FlinkJobConfigVO> getById(@PathVariable String clusterId, @PathVariable Integer id) {
        FlinkJobConfig config = flinkJobConfigService.getById(id);
        if (config == null) {
            return Result.error("配置不存在");
        }
        
        FlinkJobConfigVO vo = new FlinkJobConfigVO();
        BeanUtils.copyProperties(config, vo);
        
        // 获取当前运行的作业
        AbstractFlinkJobService flinkClient = getFlinkClient(clusterId);
        List<JobInfo> runningJobs = flinkClient.getRunningJobs();
        boolean isRunning = runningJobs.stream()
            .anyMatch(job -> job.getJobName().equals(config.getJobName()));
        
        vo.setStatus(isRunning ? 1 : 0);
        return Result.success(vo);
    }

    @PostMapping
    @Operation(summary = "新增Flink任务配置")
    @OperationLog(operationType = "CREATE", description = "创建Flink任务配置")
    public Result<Void> save(@RequestBody FlinkJobConfigDTO configDTO) {
        flinkJobConfigService.save(configDTO);
        return Result.success();
    }

    @PutMapping("/{id}")
    @Operation(summary = "修改Flink任务配置")
    @OperationLog(operationType = "UPDATE", description = "更新Flink任务配置")
    public Result<Void> update(@PathVariable Integer id, @RequestBody FlinkJobConfigDTO configDTO) {
        flinkJobConfigService.update(id, configDTO);
        return Result.success();
    }

    @DeleteMapping("/{id}")
    @Operation(summary = "删除Flink任务配置")
    @OperationLog(operationType = "DELETE", description = "删除Flink任务配置")
    public Result<Void> delete(@PathVariable Integer id) {
        flinkJobConfigService.delete(id);
        return Result.success();
    }
} 